package com.fanxl.flink.datastream.windows;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeoutTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Iterator;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/3/10 11:02
 */
public class WindowApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        test05(env);
        env.execute("WindowApp");
    }

    public static void test05(StreamExecutionEnvironment env) {
        env.setParallelism(1);
        env.socketTextStream("10.215.15.200", 9527)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String value) throws Exception {
                        System.out.println("input:" + value);
                        return Integer.parseInt(value);
                    }
                })
                .filter(new FilterFunction<Integer>() {
                    @Override
                    public boolean filter(Integer value) throws Exception {
                        return value > 7;
                    }
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .trigger(ProcessingTimeoutTrigger.of(ProcessingTimeTrigger.create(), Duration.ofSeconds(5L)))
                .process(new ProcessAllWindowFunction<Integer, String, TimeWindow>() {

                    @Override
                    public void process(Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                        System.out.println("触发了");
                        if (!elements.iterator().hasNext()) {
                            out.collect("不发生算子触发了");
                        }
                    }
                })
                .print();
    }

    /**
     * 每秒限流5条数据
     * @param env
     */
    public static void test04(StreamExecutionEnvironment env) {
        env.setParallelism(1);
        env.socketTextStream("10.215.15.200", 9527)
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new ProcessAllWindowFunction<String, String, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        int count = 0;
                        Iterator<String> iterator = elements.iterator();
                        while (iterator.hasNext()) {
                            count += 1;
                            if (count > 5){
                                break;
                            }
                            out.collect(iterator.next());
                        }
                    }
                })
                .print();
    }

    public static void test03(StreamExecutionEnvironment env) {
        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        return Tuple2.of("pk", Integer.parseInt(value));
                    }
                }).keyBy(x -> x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new PKProcessWindowFunction())
                .print();
    }


    public static void test02(StreamExecutionEnvironment env) {
        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        String[] splits = value.split(",");
                        return Tuple2.of(splits[0].trim(), Integer.parseInt(splits[1].trim()));
                    }
                }).keyBy(x -> x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {

                        System.out.println("value1 = [" + value1 + "], value2 = [" + value2 + "]");
                        return Tuple2.of(value1.f0, value1.f1+value2.f1);
                    }
                })
                .print();
    }


    public static void test01(StreamExecutionEnvironment env) {
//        env.socketTextStream("localhost", 9527)
//                .map(new MapFunction<String, Integer>() {
//                    @Override
//                    public Integer map(String value) throws Exception {
//                        return Integer.parseInt(value);
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum(0)
//                .print();

//        env.socketTextStream("localhost", 9527)
//                .map(new MapFunction<String, Integer>() {
//                    @Override
//                    public Integer map(String value) throws Exception {
//                        return Integer.parseInt(value);
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum(0)
//                .print();


        // hadoop,1  spark,1
        env.socketTextStream("localhost", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        String[] splits = value.split(",");
                        return Tuple2.of(splits[0].trim(), Integer.parseInt(splits[1].trim()));
                    }
                }).keyBy(x -> x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();
    }

}
